package org.tera;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class TG extends Configured implements Tool {

    static String rowskey = "tera.rows";
    
    /**
     * @param args
     */

    static long getNumberOfRows(JobConf job) {
        return job.getLong(rowskey, 0);
    }

    static void setNumberOfRows(JobConf job, long numRows) {
        job.setLong(rowskey, numRows);
    }
    

    @Override
    public int run(String[] args) throws Exception {
        JobConf job = (JobConf) getConf();
        setNumberOfRows(job, Long.parseLong(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.setJobName("TG");
        job.setJarByClass(TG.class);
        
        job.setMapperClass(TgMapper.class);
        
        job.setNumReduceTasks(0);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        job.setInputFormat(TgInpForm.class);
        
        JobClient.runJob(job);
        return 0;
    }
    
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new JobConf(), new TG(), args);
        System.exit(res);
      }

}
